package org.example.rocketmq.order;

import org.apache.rocketmq.client.exception.MQBrokerException;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.MessageQueueSelector;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageQueue;
import org.apache.rocketmq.remoting.exception.RemotingException;

import java.util.ArrayList;
import java.util.List;

/**
 * 顺序消息
 * 生产者向特定的队列发送消息，就能使消息的消费有序
 */
public class OrderProducer {
    public static void main(String[] args) throws InterruptedException, RemotingException, MQClientException, MQBrokerException {
        asyncSend();
    }

    public static DefaultMQProducer getProducer() {
        //1.创建生产者producer，并指定组名
        DefaultMQProducer producer = new DefaultMQProducer("producer-group1");
        //2.指定nameserver
        producer.setNamesrvAddr("192.168.3.109:9876;192.168.3.108:9876");
        return producer;
    }

    /**
     * 发送同步消息
     */
    public static void asyncSend() throws MQClientException, RemotingException, InterruptedException, MQBrokerException {
        //1.创建生产者producer，并指定组名
        DefaultMQProducer producer = getProducer();
        //3.启动producer
        producer.start();

        List<Order> orders = Order.buildOrderList();

        for (Order order : orders) {
            /**
             *  参数
             *  1. 消息对象
             *  2. 消息队列选择器
             *  3. 选择队列的业务标志
             */
            Message message = new Message("orderTopic", "order", order.toString().getBytes());
            SendResult result = producer.send(message, new MessageQueueSelector() {
                /**
                 *
                 * @param mqs   队列
                 * @param msg   消息对象
                 * @param arg   业务标志参数   send方法中的第三个参数
                 * @return
                 */
                @Override
                public MessageQueue select(List<MessageQueue> mqs, Message msg, Object arg) {
                    int orderId = (int) arg;
                    int index = orderId % mqs.size();
                    //同一个订单Id的消息发送到同一个队列中
                    return mqs.get(index);
                }
            }, order.getOrderId());

            System.out.println("发送返回结果 : " + result);
        }

        //6.关闭生产者
        producer.shutdown();
    }

    static class Order {
        private int orderId;
        private String msg;

        public Order(int orderId, String msg) {
            this.orderId = orderId;
            this.msg = msg;
        }

        public int getOrderId() {
            return orderId;
        }

        public void setOrderId(int orderId) {
            this.orderId = orderId;
        }

        public String getMsg() {
            return msg;
        }

        public void setMsg(String msg) {
            this.msg = msg;
        }

        @Override
        public String toString() {
            return "Order{" +
                    "orderId=" + orderId +
                    ", msg='" + msg + '\'' +
                    '}';
        }

        public static List<Order> buildOrderList() {
            List<Order> orders = new ArrayList<>();
            orders.add(new Order(159, "下单"));
            orders.add(new Order(179, "下单"));
            orders.add(new Order(179, "付款"));
            orders.add(new Order(159, "付款"));
            orders.add(new Order(179, "发货"));
            orders.add(new Order(159, "发货"));
            orders.add(new Order(159, "完成"));
            orders.add(new Order(179, "完成"));

            return orders;
        }
    }
}
